package com.buddy.zookeeper.master;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;

public class MasterAsync implements Watcher {

    private static final Logger LOG = LoggerFactory.getLogger(MasterAsync.class);


    ZooKeeper zk;
    String hostPort;
    String serverId = Integer.toString(new Random().nextInt());
    static boolean isLeader = false;
    AsyncCallback.StringCallback masterCreateCallback = new AsyncCallback.StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();
                    return;
                case OK:
                    isLeader = true;
                    break;
                default:
                    isLeader = false;
            }
        }

        DataCallback masterCheckCallBack = new DataCallback() {
            public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
                switch (KeeperException.Code.get(rc)){
                    case CONNECTIONLOSS:
                        checkMaster();
                        return;
                    case NONODE:
                        runForMaster();
                        return;
                }
            }
        };
        void checkMaster() {
            zk.getData("/master",false,masterCheckCallBack,null);
        }

    };

    MasterAsync(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort, 15000, this);
    }

    void stopZK() throws InterruptedException {
        zk.close();
    }


    void runForMaster() {
        zk.create("/master",
                serverId.getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL,
                masterCreateCallback,
                null);
    }

    public void process(WatchedEvent watchedEvent) {
        System.out.println(watchedEvent);
    }


    public void bootstrap(){
        createParent("/workers",new byte[0]);
        createParent("/assign",new byte[0]);
        createParent("/tasks",new byte[0]);
    }

    private void createParent(String path, byte[] data) {
        zk.create(path,data, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT,createParentCallback,data);
    }
    AsyncCallback.StringCallback createParentCallback = new AsyncCallback.StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)){
                case CONNECTIONLOSS:
                    createParent(path, (byte[]) ctx);
                    break;
                case OK:
                    LOG.info("Parent created!");
                    break;
                case NODEEXISTS:
                    LOG.warn("Parent already registered:"+path);
                    break;
                default:
                    LOG.error("Something went wrong:",KeeperException.create(KeeperException.Code.get(rc),path));
            }
        }
    };

    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
        MasterAsync master = new MasterAsync("127.0.0.1:2181");
        master.startZK();
//        master.runForMaster();
//        while (!isLeader){
//            System.out.println(isLeader);
//            Thread.sleep(100);
//        }
//        System.out.println("------------>I'm the leader!");
        master.bootstrap();// create parent znode
        master.stopZK();
    }
}
